package com.jetlinks.protocol;

import com.alibaba.fastjson.JSONObject;
import com.jetlinks.protocol.binary.BinaryMessageType;
import com.jetlinks.protocol.handle.MyMsgHandle;
import com.jetlinks.protocol.handle.MyMsgHandleImpl;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.ByteBufUtil;
import io.netty.buffer.Unpooled;
import lombok.extern.slf4j.Slf4j;
import org.jetlinks.core.message.*;
import org.jetlinks.core.message.codec.*;
import org.jetlinks.core.message.function.FunctionInvokeMessage;
import org.jetlinks.core.message.function.FunctionParameter;
import org.jetlinks.core.message.property.ReadPropertyMessage;
import org.jetlinks.core.message.property.ReadPropertyMessageReply;
import org.jetlinks.core.message.property.ReportPropertyMessage;
import org.jetlinks.core.server.session.DeviceSession;

import java.io.UnsupportedEncodingException;
import java.time.LocalDateTime;

import org.reactivestreams.Publisher;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;

import javax.annotation.Resource;
import java.time.ZoneOffset;
import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author Huluw
 */
@Slf4j
public class PowerDeviceMessageCodec implements DeviceMessageCodec {
    private MyMsgHandleImpl myMsgHandle = new MyMsgHandleImpl();

    // 传输协议定义
    @Override
    public Transport getSupportTransport() {
        return DefaultTransport.TCP;
    }

    // 把tcp消息解码为平台消息，多用于设备上报消息到平台
    @Override
    public Flux<? extends Message> decode(MessageDecodeContext context) {
        return Flux.defer(() -> {
            // 消息上下文
            FromDeviceMessageContext ctx = ((FromDeviceMessageContext) context);
            // 从上下文中获取消息字节数组
            ByteBuf byteBuf = context.getMessage().getPayload();
            byte[] payload = ByteBufUtil.getBytes(byteBuf, 0, byteBuf.readableBytes(), false);
            // 把字节流转换为字符串，根据不同设备不同协议进行解析，
            String text = null;
            try {
                text = new String(payload, "UTF-8");
            } catch (UnsupportedEncodingException e) {
                return Mono.empty();
            }
            String[] items = text.split("\\x01");    //消息格式  8=HC1.09=66135=034=85115
            JSONObject jsonObject = myMsgHandle.converItems(items);

            //获取上报属性
            Map<String, Object> properties = myMsgHandle.getMap(jsonObject);
            ReportPropertyMessage message = new ReportPropertyMessage();
            // 设置消息ID
            message.setMessageId("2");
            //设置消息设备号
            message.setDeviceId("1");
            // 以当前时间戳为消息时间
            long time = LocalDateTime.now().toInstant(ZoneOffset.of("+8")).toEpochMilli();
            message.setTimestamp(time);
            // 设置上报属性
            message.setProperties(properties);
            // 获取设备会话信息
            DeviceSession session = ctx.getSession();
            // 如果session中没有设备信息，则为设备首次上线
            if (session.getOperator() == null) {
                DeviceOnlineMessage onlineMessage = new DeviceOnlineMessage();
                onlineMessage.setDeviceId("1");
                onlineMessage.setTimestamp(System.currentTimeMillis());
                // 返回到平台上线消息
                return Flux.just(message, onlineMessage);
            }
            // 返回到平台属性上报消息
//            return Mono.just(message);
            byteBuf.readInt();
            return Mono.justOrEmpty(BinaryMessageType.read(byteBuf, context.getDevice().getDeviceId()));
        });
    }

    // 把平台消息编码为协议传输消息，多用于平台命令下发到设备
    @Override
    public Publisher<? extends EncodedMessage> encode(MessageEncodeContext context) {
        // 从平台消息上下文中获取消息内容
        DeviceMessage message = (DeviceMessage) context.getMessage();
//            EncodedMessage encodedMessage = EncodedMessage.simple(Unpooled.wrappedBuffer(message.toString().getBytes()));
        // 根据消息类型的不同，构造不同的消息

        if (message instanceof ReadPropertyMessage) {
            ReadPropertyMessage readPropertyMessage = (ReadPropertyMessage) message;
            // 获取需要传输的字节
            byte[] bytes = readPropertyMessage.toString().getBytes();
            // 构造为平台传输到设备的消息体
//                encodedMessage = EncodedMessage.simple(Unpooled.wrappedBuffer(bytes));

            return Mono.just(EncodedMessage.simple(
                    BinaryMessageType.write(readPropertyMessage, Unpooled.buffer())
            ));
        }
        EncodedMessage encodedMessage = EncodedMessage.simple(Unpooled.wrappedBuffer(message.toString().getBytes()));

        if (message instanceof FunctionInvokeMessage) {
            FunctionInvokeMessage functionInvokeMessage = (FunctionInvokeMessage) message;
            String msg = functionInvokeMessage.toString();
//                String sentToDevMsg = myMsgHandle.sendToDev(msg);
            encodedMessage = EncodedMessage.simple(Unpooled.wrappedBuffer(msg.getBytes()));
        }
        return Mono.just(encodedMessage);
    }

    public static ByteBuf wrapByteByf(ByteBuf payload) {

        return Unpooled.wrappedBuffer(
            Unpooled.buffer().writeInt(payload.writerIndex()),
            payload);
    }


}
